package com.lu.mqtt

import android.app.Application
import android.util.Log
import org.eclipse.paho.android.service.MqttAndroidClient
import org.eclipse.paho.client.mqttv3.*
import com.lu.mqtt.listener.IMqttConnectionListener
import com.lu.mqtt.listener.IMqttMessageDispatcher
import com.lu.mqtt.tools.TrustAllManager
import java.security.SecureRandom
import java.util.concurrent.Executors
import javax.net.ssl.SSLContext

/**
 *author: luqihua
 *date:2021/10/19
 *description:
 **/
class RJMqttClient @Throws(MqttException::class) constructor(
        application: Application,
        serverURI: String,
        username: String,
        password: String,
        clientId: String,
        subTopics: List<String>) : MqttCallback {
    companion object {
        private const val TAG = "MQTT"
        private const val HEARTBEAT_INTERVAL = 20
        private const val QOS = 2
        private fun getDefaultOption() =
            MqttConnectOptions().also {
                it.isCleanSession = true //设置是否清除缓存
                it.connectionTimeout = 10 //设置超时时间，单位：秒
                it.keepAliveInterval = HEARTBEAT_INTERVAL //设置心跳包发送间隔，单位：秒
                it.mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1
                //                it.setWill(TOPIC, LAST_WILL.toByteArray(), QOS, false)
            }
    }

    private var iMessageDispatcher: IMqttMessageDispatcher? = null
    private var iMqttConnectionListener: IMqttConnectionListener? = null

    private val mMqttConnectOptions: MqttConnectOptions

    private val mMqttAndroidClient: MqttAndroidClient

    private val mThreadPoolExecutor = Executors.newCachedThreadPool()

    private val mSubTopics = ArrayList<String>()

    init {
        Log.d(TAG, "init : serverURI=$serverURI  username=$username subTopics=$subTopics")
        this.mSubTopics.addAll(subTopics)
        mMqttAndroidClient = MqttAndroidClient(application, serverURI, clientId)
        mMqttAndroidClient.setCallback(this)
        mMqttConnectOptions = getDefaultOption().also {
            it.isAutomaticReconnect = false
            it.userName = username //设置用户名
            it.password = password.toCharArray() //设置密码
            //             正式环境
            //            sslContext.init(null, null, null);
            //            SSLSocketFactory factory = sslContext.getSocketFactory();
            //            mqttConnectOptions.setSocketFactory(factory);
            //             正式环境 END
            //             测试环境
            //             说明：由于测试环境的证书不是受信的证书。所以必须自定义TrustManager将证书添加至信任或信任所有（生产环境不建议）。
            //             另外由于测试环境没有域名，还需要要禁用verifyHost。否则无法连接。
            //            sslContext.init(null, null, null)
            val sslContext = SSLContext.getInstance("SSL")
            sslContext.init(null, arrayOf(TrustAllManager()), SecureRandom())
            it.socketFactory = sslContext.socketFactory
        }
    }

    @Throws(MqttException::class)
    fun connect() {
        Log.e(
            TAG,
              "MQTT connect start: thread=${Thread.currentThread().name} hashCode=${hashCode()}")
        mMqttAndroidClient.connect(mMqttConnectOptions, null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken) {
                Log.e(TAG, "MQTT connect success ${Thread.currentThread().name} ${this.hashCode()}")
                mSubTopics.forEach {
                    mMqttAndroidClient.subscribe(it, QOS)
                }
                //连接成功，将未发送的消息发送出去
                iMqttConnectionListener?.onMqttConnected(this@RJMqttClient)
            }

            override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {
                Log.e(TAG, "MQTT connect failure: ", exception)
                iMqttConnectionListener?.onMqttDisconnected()
            }
        })
    }

    fun disconnect() {
        mMqttAndroidClient.setCallback(null)
        try {
            mMqttAndroidClient.disconnect()
        } catch (e: MqttException) {
            Log.e(TAG, "disconnect: 忽略该异常", e)
        }
    }

    fun setMessageDispatcher(dispatcher: IMqttMessageDispatcher) {
        this.iMessageDispatcher = dispatcher
    }

    fun setConnectionListener(listener: IMqttConnectionListener) {
        this.iMqttConnectionListener = listener
    }


    override fun connectionLost(cause: Throwable?) {
        Log.e(TAG, "链接丢失: ", cause)
        iMqttConnectionListener?.onMqttDisconnected()
    }

    override fun messageArrived(topic: String, message: MqttMessage) {
        val msg = String(message.payload)
        if (mSubTopics.contains(topic)) {
            mThreadPoolExecutor.execute {
                Log.w(TAG, "messageArrived  topic=$topic : $msg")
                iMessageDispatcher?.dispatch(this, topic, msg)
            }
        } else {
            Log.w(TAG, "messageArrived : unknown topic")
        }
    }

    override fun deliveryComplete(token: IMqttDeliveryToken?) {
        Log.d(TAG, "deliveryComplete : ${token?.messageId}")
    }

    /**
     * 订阅消息
     *
     * @param topic
     * @param qos
     */
    fun subscribe(topic: String, qos: Int) {
        try {
            mSubTopics.add(topic)
            mMqttAndroidClient.subscribe(topic, qos) //订阅主题，参数：主题、服务质量
        } catch (e: MqttException) {
            Log.e(TAG, "subscribe: failed", e)
        }
    }

    /**
     * Publish
     *
     * @param topic 主题
     * @param message 消息
     */
    fun publish(topic: String, message: String) {
        if (mMqttAndroidClient.isConnected) {
            doPublish(topic, message)
        } else {
            Log.d(TAG, "publish failed: disconnected")
        }
    }

    private fun doPublish(topic: String, message: String) {
        try {
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            val token = mMqttAndroidClient.publish(
                    topic,
                    message.toByteArray(),
                    QOS,
                    false,
                    null,
                    object : IMqttActionListener {
                        override fun onSuccess(asyncActionToken: IMqttToken?) {
                            Log.d(TAG, "MQTT publish success: $topic-$message")
                        }

                        override fun onFailure(
                                asyncActionToken: IMqttToken?,
                                exception: Throwable?
                        ) {
                            Log.e(TAG, "MQTT publish failure: ", exception)
                        }

                    })

            Log.d(TAG, "doPublish :$topic  ${token.messageId}")
        } catch (e: MqttException) {
            Log.e(TAG, "doPublish error : $message", e)
            e.printStackTrace()
        }
    }
}